package top.suven.base.core.kafka;

import top.suven.base.util.utilize.JsonUtils;
import com.alibaba.dubbo.common.utils.ReflectUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.util.Optional;


public abstract class KafkaConsumerAbstractHandler<T>{

    private Logger logger = LoggerFactory.getLogger(KafkaConsumerAbstractHandler.class);



    public abstract void executeConsumer(T value );

    private void setLogger(Class clazz){
        logger.warn(" abstract  executeConsumer before setLogger clazz=:[{}]" , clazz);
    }

    public void  onMessage(ConsumerRecord<String, String> record,Class<T> clazz ) {
        try{
            Optional<?> messages = Optional.ofNullable(record.value());
            if (!messages.isPresent()) {
                return;
            }
            if(clazz == Object.class|| null == clazz){
                clazz = this.getTempalteType();
            }
            logger.info("  onMessage ConsumerRecord  value=:[{}]" , record.value());
            setLogger(clazz);
            if(null == clazz){
                clazz = (Class<T>) String.class;
            }
            if(String.class == clazz || String.class == clazz || clazz == Object.class ){
                executeConsumer((T)record.value());
            }else {
                T instance = JsonUtils.readValue(record.value(),clazz);
                executeConsumer(instance);
            }

        }catch (Exception e){
            e.printStackTrace();
            logger.warn(" abstract kafka onMessage  Exception clazz=:[{}] ,ConsumerRecord =:[{}] " , clazz,record.toString());
        }


    }

    private static Method getMethodByName(Class<?> clazz, String methodName) {
        try {
            return ReflectUtils.findMethodByMethodName(clazz, methodName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    private Class<T> getTempalteType() {
        Class<T> clazz = (Class<T>) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        return clazz;
    }



}
